package com.myorg.bionline.gui.control;

import java.io.IOException;
import java.io.Serializable;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.faces.bean.ManagedBean;
import javax.faces.bean.ViewScoped;

import org.apache.http.client.config.RequestConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.myorg.bionline.driver.WebSocketClient;
import com.myorg.domain.BlockInfo;
import com.myorg.domain.SearchRequest;
import com.myorg.domain.SearchResult;
import com.myorg.search.BaseBean;
import com.myorg.util.ConstantsToBeExternalized;
import com.myorg.util.DataLocalityRunner;
import com.myorg.util.DataLocalityRunner.DataLocalityTask;
import com.myorg.util.HdfsHelper;

@ManagedBean
@ViewScoped
public class TestBean extends BaseBean implements Serializable {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1066280761970434744L;
	final static private Logger logger = LoggerFactory.getLogger(TestBean.class);

	public void doSomething() {
		logger.info("I am doing something now...");

		WebSocketClient wsClient = new WebSocketClient(URI.create("ws://localhost:8080/worker/"));

		wsClient.send("hi websocket server, I am here!");
		wsClient.close();
	}

	public SearchResult dispatchSearch(WebSocketClient wsClient, final SearchRequest request, final String contextPath,
			final int port, final String syncPath) {
		RequestConfig requestConfig = RequestConfig.custom()
				.setSocketTimeout(ConstantsToBeExternalized.DEFAULT_SOCKET_TIMEOUT)
				.setConnectTimeout(ConstantsToBeExternalized.DEFAULT_CONNECTION_TIMEOUT).build();
		final SearchResult rtn = new SearchResult();
		try {
			List<BlockInfo> blockInfos = generateBlockInfoList(request.getFilePaths());
			dumpBlockInfo(blockInfos);

			final CountDownLatch latch = new CountDownLatch(blockInfos.size());

			ExecutorService executor = Executors.newCachedThreadPool();

			Map<BlockInfo, String> map = new HashMap<BlockInfo, String>();
			if (blockInfos != null) {
				for (BlockInfo blkInfo : blockInfos) {
					String host = selectHost(blkInfo, map);
					map.put(blkInfo, host);
				}

				for (BlockInfo blkInfo : map.keySet()) {
					String host = map.get(blkInfo);

					Runnable r = new MyRunnable(host, blkInfo, request);
					executor.submit(r);
				}
			}

			DataLocalityRunner.runWithLocality(blockInfos, new DataLocalityTask() {
				@Override
				public void run(BlockInfo blkInfo, String selectedHost) {
				}
			});

			latch.await();
			logger.info("completed");

			return rtn.max(request.getMax());
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}

	private static List<BlockInfo> generateBlockInfoList(String[] filePaths) throws IOException {
		HdfsHelper helper = new HdfsHelper();
		List<BlockInfo> rtn = new ArrayList<BlockInfo>();
		for (String path : filePaths) {
			rtn.addAll(helper.getBlockInfos(path));
		}
		return rtn;
	}

	private static void dumpBlockInfo(List<BlockInfo> blkInfos) throws IOException {
		for (int i = 0; i < blkInfos.size(); i++) {
			String[] hosts = blkInfos.get(i).getBlockLocation().getHosts();
			String[] pathInfo = blkInfos.get(i).getBlockLocation().getTopologyPaths();
			for (int j = 0; j < pathInfo.length; j++) {
				logger.info("getTopologyPaths[" + i + "]" + pathInfo[j].toString());
			}
			logger.info("Host %d: %s %n", i, hosts);
		}
	}

	private static String selectHost(BlockInfo blkInfo, Map<BlockInfo, String> map) {
		int score = -1;
		String selected = null;
		try {
			for (String host : blkInfo.getBlockLocation().getHosts()) {
				if (selected == null) {
					selected = host;
					// score = getScore(host, map);
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
		return selected;
	}

}
